package com.gz.rocketmq.listen;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

@Component
public class RocketListen{

    @Resource
    private DefaultListen defaultListen;

    @Value("${rocketmq.consumer.nameServer}")
    private String nameSerAdd;

    @Value("${rocketmq.consumer.group}")
    private String group;

    @PostConstruct
    public void listenDefaultConsumer(){
        try {
            defaultConsumer("springboot","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public DefaultMQPushConsumer defaultConsumer(String topic,String tag) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameSerAdd);
        consumer.subscribe(topic,tag);
        consumer.registerMessageListener(defaultListen);
        consumer.start();
        System.out.println("=====启动=====");
        return consumer;
    }
}
